package cas.ihep.hep.convert;

import cas.ihep.fs.UnifiedBlock;
import cas.ihep.fs.UnifiedFile;
import cas.ihep.fs.UnifiedFileSystem;
import cas.ihep.hep.config.Config;
import scala.Tuple2;

import java.io.*;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

public class Root2Alluxio {
    public String basedir;

    public Root2Alluxio(){
        basedir="/tmp/"+Long.toHexString(System.currentTimeMillis())+'-'+Integer.toHexString(new Random().nextInt());
    }

    private void processLocal(BufferedReader reader, String filename, UnifiedFileSystem ufs, List<String> outfiles)throws IOException,URISyntaxException{
        String line;
        int i=0;
        while((line=reader.readLine())!=null){
            String curfile=filename+i++;
            try(UnifiedFile cur=ufs.create(new URI(curfile))){
                UnifiedBlock blk=cur.next();
                File dst=new File(blk.absolutePath());
                //noinspection ResultOfMethodCallIgnored
                new File(line).renameTo(dst);
                outfiles.add(curfile);
            }
        }
    }

    static class FileChannelInputStream extends InputStream{

        FileChannelInputStream(File file_,boolean deleteonclose)throws IOException{
            file=file_;
            channel=FileChannel.open(file.toPath(),StandardOpenOption.READ);
            buffer=channel.map(FileChannel.MapMode.READ_ONLY,0,file.length());
            autodelete=deleteonclose;
        }

        File file;
        FileChannel channel;
        MappedByteBuffer buffer;
        boolean autodelete;

        @Override
        public int read() {
            return buffer.get();
        }

        public int read(byte[] buf,int off,int len){
            int remain=buffer.remaining();
            if(remain<len){
                len=remain;
            }else if(remain==0){
                return -1;
            }
            buffer.get(buf,off,len);
            return len;
        }

        public int read(byte[] buf){
            int remain=buffer.remaining();
            if(remain>buf.length){
                remain=buf.length;
            }else if(remain==0){
                return -1;
            }
            buffer.get(buf,0,remain);
            return remain;
        }

        public int available(){
            return buffer.remaining();
        }

        public long skip(long sz){
            int cur=buffer.position();
            buffer.position((int) (cur+sz));
            return buffer.remaining();
        }

        @Override
        public void close() throws IOException {
            try {
                Method cleaner=buffer.getClass().getMethod("cleaner");
                cleaner.setAccessible(true);
                Object cleaner_object=cleaner.invoke(buffer);
                Method clean=cleaner_object.getClass().getMethod("clean");
                clean.setAccessible(true);
                clean.invoke(cleaner_object);
            }catch (NoSuchMethodException|IllegalAccessException | InvocationTargetException ex){
                throw new IOException(ex);
            }
            buffer=null;
            channel.close();
            if (autodelete){
                file.delete();
            }
        }
    }

    private void processRemote(BufferedReader reader, String filename, UnifiedFileSystem afs,List<String> outfiles) throws IOException, URISyntaxException {
        String line;
        int i=0;
        byte[] buf=new byte[8192];
        while((line=reader.readLine())!=null){
            String curfile=filename+i++;
            File infile=new File(line);
            try(OutputStream os=afs.createStream(new URI(curfile));
                FileChannelInputStream is=new FileChannelInputStream(infile,true)){
                int each;
                while((each=is.read(buf))>0){
                    os.write(buf,0,each);
                }
                outfiles.add(curfile);
            }
        }
    }

    public Tuple2<List<String>,Long> convert(UnifiedFileSystem ufs,Config config) throws Exception {
        ArrayList<String> cmdlist=new ArrayList<>();
        String filename=config.getString("filename");
        cmdlist.add(config.getString("convertor")); cmdlist.add(filename); cmdlist.add(config.getString("treename"));
        cmdlist.add(Integer.toString(config.getInt("events"))); cmdlist.add("20000"); cmdlist.addAll(config.getStringArray("branch"));
        ProcessBuilder builder=new ProcessBuilder();
        builder.redirectError(ProcessBuilder.Redirect.INHERIT);
        builder.command(cmdlist);
        Process subprocess=builder.start();
        cmdlist.clear();
        Tuple2<List<String>,Long> ret;
        filename=basedir+'/'+filename+'/';
        try(BufferedReader reader=new BufferedReader(new InputStreamReader(subprocess.getInputStream()))){
            long entries=Long.parseLong(reader.readLine());
            if(ufs.localWorker("alluxio")){
                processLocal(reader,filename,ufs,cmdlist);
            }else{
                processRemote(reader,filename,ufs,cmdlist);
            }
            ret=new Tuple2<>(cmdlist,entries);
        }
        if (subprocess.waitFor()!=0){
            throw new RuntimeException("Exit error");
        }
        subprocess.destroy();
        return ret;
    }
}
